/*
 * Copyright (C) 2006 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.google.common.util.concurrent;

import static com.google.common.base.Preconditions.checkNotNull;

import java.lang.reflect.UndeclaredThrowableException;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.annotation.Nullable;

import com.google.common.base.Function;

/**
 * Static utility methods pertaining to the {@link Future} interface.
 * 
 * @author Kevin Bourrillion
 * @author Nishant Thakkar
 * @author Sven Mawson
 * @since 2009.09.15 <b>tentative</b>
 */
public class Futures {
    private Futures() {
    }

    /**
     * Returns an uninterruptible view of a {@code Future}. If a thread is
     * interrupted during an attempt to {@code get()} from the returned future,
     * it continues to wait on the result until it is available or the timeout
     * elapses, and only then re-interrupts the thread.
     */
    public static <V> UninterruptibleFuture<V> makeUninterruptible(
            final Future<V> future) {
        checkNotNull(future);
        if (future instanceof UninterruptibleFuture) {
            return (UninterruptibleFuture<V>) future;
        }
        return new UninterruptibleFuture<V>() {
            public boolean cancel(boolean mayInterruptIfRunning) {
                return future.cancel(mayInterruptIfRunning);
            }

            public boolean isCancelled() {
                return future.isCancelled();
            }

            public boolean isDone() {
                return future.isDone();
            }

            public V get(long timeoutDuration, TimeUnit timeoutUnit)
                    throws TimeoutException, ExecutionException {
                boolean interrupted = false;
                try {
                    long timeoutNanos = timeoutUnit.toNanos(timeoutDuration);
                    long end = System.nanoTime() + timeoutNanos;
                    for (long remaining = timeoutNanos; remaining > 0; remaining = end
                            - System.nanoTime()) {
                        try {
                            return future.get(remaining, TimeUnit.NANOSECONDS);
                        } catch (InterruptedException ignored) {
                            interrupted = true;
                        }
                    }
                    throw new TimeoutException();
                } finally {
                    if (interrupted) {
                        Thread.currentThread().interrupt();
                    }
                }
            }

            public V get() throws ExecutionException {
                boolean interrupted = false;
                try {
                    while (true) {
                        try {
                            return future.get();
                        } catch (InterruptedException ignored) {
                            interrupted = true;
                        }
                    }
                } finally {
                    if (interrupted) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
        };
    }

    /**
     * Creates a {@link ListenableFuture} out of a normal {@link Future}. The
     * returned future will create a thread to wait for the source future to
     * complete before executing the listeners.
     * 
     * <p>
     * Callers who have a future that subclasses
     * {@link java.util.concurrent.FutureTask} may want to instead subclass
     * {@link ListenableFutureTask}, which adds the {@link ListenableFuture}
     * functionality to the standard {@code FutureTask} implementation.
     */
    public static <T> ListenableFuture<T> makeListenable(Future<T> future) {
        if (future instanceof ListenableFuture) {
            return (ListenableFuture<T>) future;
        }
        return new ListenableFutureAdapter<T>(future);
    }

    /**
     * Creates a {@link CheckedFuture} out of a normal {@link Future} and a
     * {@link Function} that maps from {@link Exception} instances into the
     * appropriate checked type.
     * 
     * <p>
     * The given mapping function will be applied to an
     * {@link InterruptedException}, a {@link CancellationException}, or an
     * {@link ExecutionException} with the actual cause of the exception. See
     * {@link Future#get()} for details on the exceptions thrown.
     */
    public static <T, E extends Exception> CheckedFuture<T, E> makeChecked(
            Future<T> future, Function<Exception, E> mapper) {
        return new MappingCheckedFuture<T, E>(makeListenable(future), mapper);
    }

    /**
     * Creates a {@code ListenableFuture} which has its value set immediately
     * upon construction. The getters just return the value. This {@code Future}
     * can't be canceled or timed out and its {@code isDone()} method always
     * returns {@code true}. It's useful for returning something that implements
     * the {@code ListenableFuture} interface but already has the result.
     */
    public static <T> ListenableFuture<T> immediateFuture(@Nullable T value) {
        ValueFuture<T> future = ValueFuture.create();
        future.set(value);
        return future;
    }

    /**
     * Creates a {@code CheckedFuture} which has its value set immediately upon
     * construction. The getters just return the value. This {@code Future} 
     * can't be canceled or timed out and its {@code isDone()} method always
     * returns {@code true}. It's useful for returning something that implements
     * the {@code CheckedFuture} interface but already has the result.
     */
    public static <T, E extends Exception> CheckedFuture<T, E> immediateCheckedFuture(
            @Nullable T value) {
        ValueFuture<T> future = ValueFuture.create();
        future.set(value);
        return Futures.makeChecked(future, new Function<Exception, E>() {
            public E apply(Exception e) {
                throw new AssertionError("impossible");
            }
        });
    }

    /**
     * Creates a {@code ListenableFuture} which has an exception set immediately
     * upon construction. The getters just return the value. This {@code Future}
     * can't be canceled or timed out and its {@code isDone()} method always
     * returns {@code true}. It's useful for returning something that implements
     * the {@code ListenableFuture} interface but already has a failed result.
     * Calling {@code get()} will throw the provided {@code Throwable} (wrapped
     * in an {@code ExecutionException}).
     * 
     * @throws Error
     *             if the throwable was an {@link Error}.
     */
    public static <T> ListenableFuture<T> immediateFailedFuture(
            Throwable throwable) {
        checkNotNull(throwable);
        ValueFuture<T> future = ValueFuture.create();
        future.setException(throwable);
        return future;
    }

    /**
     * Creates a {@code CheckedFuture} which has an exception set immediately
     * upon construction. The getters just return the value. This {@code Future}
     * can't be canceled or timed out and its {@code isDone()} method always
     * returns {@code true}. It's useful for returning something that implements
     * the {@code CheckedFuture} interface but already has a failed result.
     * Calling {@code get()} will throw the provided {@code Throwable} (wrapped
     * in an {@code ExecutionException}) and calling {@code checkedGet()} will
     * throw the provided exception itself.
     * 
     * @throws Error
     *             if the throwable was an {@link Error}.
     */
    public static <T, E extends Exception> CheckedFuture<T, E> immediateFailedCheckedFuture(
            final E exception) {
        checkNotNull(exception);
        return makeChecked(Futures.<T> immediateFailedFuture(exception),
                new Function<Exception, E>() {
                    public E apply(Exception e) {
                        return exception;
                    }
                });
    }

    /**
     * Creates a new {@code ListenableFuture} that wraps another
     * {@code ListenableFuture}. The result of the new future is the result of
     * the provided function called on the result of the provided future. The
     * resulting future doesn't interrupt when aborted.
     * 
     * <p>
     * TODO: Add a version that accepts a normal {@code Future}
     * 
     * <p>
     * The typical use for this method would be when a RPC call is dependent on
     * the results of another RPC. One would call the first RPC (input), create
     * a function that calls another RPC based on input's result, and then call
     * chain on input and that function to get a {@code ListenableFuture} of the
     * result.
     * 
     * @param input
     *            The future to chain
     * @param function
     *            A function to chain the results of the provided future to the
     *            results of the returned future. This will be run in the thread
     *            that notifies input it is complete.
     * @return A future that holds result of the chain.
     */
    public static <I, O> ListenableFuture<O> chain(
            ListenableFuture<I> input,
            Function<? super I, ? extends ListenableFuture<? extends O>> function) {
        return chain(input, function, Executors.sameThreadExecutor());
    }

    /**
     * Creates a new {@code ListenableFuture} that wraps another
     * {@code ListenableFuture}. The result of the new future is the result of
     * the provided function called on the result of the provided future. The
     * resulting future doesn't interrupt when aborted.
     * 
     * <p>
     * This version allows an arbitrary executor to be passed in for running the
     * chained Function. When using {@link Executors#sameThreadExecutor}, the
     * thread chained Function executes in will be whichever thread set the
     * result of the input Future, which may be the network thread in the case
     * of RPC-based Futures.
     * 
     * @param input
     *            The future to chain
     * @param function
     *            A function to chain the results of the provided future to the
     *            results of the returned future.
     * @param exec
     *            Executor to run the function in.
     * @return A future that holds result of the chain.
     */
    public static <I, O> ListenableFuture<O> chain(
            ListenableFuture<I> input,
            Function<? super I, ? extends ListenableFuture<? extends O>> function,
            Executor exec) {
        ChainingListenableFuture<I, O> chain = new ChainingListenableFuture<I, O>(
                function, input);
        input.addListener(chain, exec);
        return chain;
    }

    /**
     * Creates a new {@code ListenableFuture} that wraps another
     * {@code ListenableFuture}. The result of the new future is the result of
     * the provided function called on the result of the provided future. The
     * resulting future doesn't interrupt when aborted.
     * 
     * <p>
     * An example use of this method is to convert a serializable object
     * returned from an RPC into a POJO.
     * 
     * @param future
     *            The future to compose
     * @param function
     *            A Function to compose the results of the provided future to
     *            the results of the returned future. This will be run in the
     *            thread that notifies input it is complete.
     * @return A future that holds result of the composition.
     */
    public static <I, O> ListenableFuture<O> compose(
            ListenableFuture<I> future,
            final Function<? super I, ? extends O> function) {
        return compose(future, function, Executors.sameThreadExecutor());
    }

    /**
     * Creates a new {@code ListenableFuture} that wraps another
     * {@code ListenableFuture}. The result of the new future is the result of
     * the provided function called on the result of the provided future. The
     * resulting future doesn't interrupt when aborted.
     * 
     * <p>
     * An example use of this method is to convert a serializable object
     * returned from an RPC into a POJO.
     * 
     * <p>
     * This version allows an arbitrary executor to be passed in for running the
     * chained Function. When using {@link Executors#sameThreadExecutor}, the
     * thread chained Function executes in will be whichever thread set the
     * result of the input Future, which may be the network thread in the case
     * of RPC-based Futures.
     * 
     * @param future
     *            The future to compose
     * @param function
     *            A Function to compose the results of the provided future to
     *            the results of the returned future.
     * @param exec
     *            Executor to run the function in.
     * @return A future that holds result of the composition.
     * @since 2010.01.04 <b>tentative</b>
     */
    public static <I, O> ListenableFuture<O> compose(
            ListenableFuture<I> future,
            final Function<? super I, ? extends O> function, Executor exec) {
        Function<I, ListenableFuture<O>> wrapperFunction = new Function<I, ListenableFuture<O>>() {
            /* @Override */public ListenableFuture<O> apply(I input) {
                O output = function.apply(input);
                return immediateFuture(output);
            }
        };
        return chain(future, wrapperFunction, exec);
    }

    /**
     * Creates a new {@code Future} that wraps another {@code Future}. The
     * result of the new future is the result of the provided function called on
     * the result of the provided future.
     * 
     * <p>
     * An example use of this method is to convert a Future that produces a
     * handle to an object to a future that produces the object itself.
     * 
     * <p>
     * Each call to {@code Future<O>.get(*)} results in a call to
     * {@code Future<I>.get(*)}, but {@code function} is only applied once, so
     * it is assumed that {@code Future<I>.get(*)} is idempotent.
     * 
     * <p>
     * When calling {@link Future#get(long, TimeUnit)} on the returned future,
     * the timeout only applies to the future passed in to this method. Any
     * additional time taken by applying {@code function} is not considered.
     * 
     * @param future
     *            The future to compose
     * @param function
     *            A Function to compose the results of the provided future to
     *            the results of the returned future. This will be run in the
     *            thread that calls one of the varieties of {@code get()}.
     * @return A future that computes result of the composition.
     */
    public static <I, O> Future<O> compose(final Future<I> future,
            final Function<? super I, ? extends O> function) {

        return new Future<O>() {

            /*
             * Concurrency detail:
             * 
             * <p>To preserve the idempotency of calls to this.get(*) calls to
             * the function are only applied once. A lock is required to prevent
             * multiple applications of the function. The calls to future.get(*)
             * are performed outside the lock, as is required to prevent calls
             * to get(long, TimeUnit) to persist beyond their timeout.
             * 
             * <p>Calls to future.get(*) on every call to this.get(*) also
             * provide the cancellation behavior for this.
             * 
             * <p>(Consider: in thread A, call get(), in thread B call get(long,
             * TimeUnit). Thread B may have to wait for Thread A to finish,
             * which would be unacceptable.)
             * 
             * <p>Note that each call to Future<O>.get(*) results in a call to
             * Future<I>.get(*), but the function is only applied once, so
             * Future<I>.get(*) is assumed to be idempotent.
             */

            private final Object lock = new Object();
            private boolean set = false;
            private O value = null;

            /* @Override */
            public O get() throws InterruptedException, ExecutionException {
                return apply(future.get());
            }

            /* @Override */
            public O get(long timeout, TimeUnit unit)
                    throws InterruptedException, ExecutionException,
                    TimeoutException {
                return apply(future.get(timeout, unit));
            }

            private O apply(I raw) {
                synchronized (lock) {
                    if (!set) {
                        value = function.apply(raw);
                        set = true;
                    }
                    return value;
                }
            }

            /* @Override */
            public boolean cancel(boolean mayInterruptIfRunning) {
                return future.cancel(mayInterruptIfRunning);
            }

            /* @Override */
            public boolean isCancelled() {
                return future.isCancelled();
            }

            /* @Override */
            public boolean isDone() {
                return future.isDone();
            }
        };
    }

    /**
     * An implementation of {@code ListenableFuture} that also implements
     * {@code Runnable} so that it can be used to nest ListenableFutures. Once
     * the passed-in {@code ListenableFuture} is complete, it calls the
     * passed-in {@code Function} to generate the result. The resulting future
     * doesn't interrupt when aborted.
     * 
     * <p>
     * If the function throws any checked exceptions, they should be wrapped in
     * a {@code UndeclaredThrowableException} so that this class can get access
     * to the cause.
     */
    private static class ChainingListenableFuture<I, O> extends
            AbstractListenableFuture<O> implements Runnable {

        private final Function<? super I, ? extends ListenableFuture<? extends O>> function;
        private final UninterruptibleFuture<? extends I> inputFuture;

        private ChainingListenableFuture(
                Function<? super I, ? extends ListenableFuture<? extends O>> function,
                ListenableFuture<? extends I> inputFuture) {
            this.function = function;
            this.inputFuture = makeUninterruptible(inputFuture);
        }

        public void run() {
            try {
                I sourceResult;
                try {
                    sourceResult = inputFuture.get();
                } catch (CancellationException e) {
                    // Cancel this future and return.
                    cancel();
                    return;
                } catch (ExecutionException e) {
                    // Set the cause of the exception as this future's exception
                    setException(e.getCause());
                    return;
                }

                final ListenableFuture<? extends O> outputFuture = function
                        .apply(sourceResult);
                outputFuture.addListener(new Runnable() {
                    public void run() {
                        try {
                            // Here it would have been nice to have had an
                            // UninterruptibleListenableFuture, but we don't
                            // want to start a
                            // combinatorial explosion of interfaces, so we have
                            // to make do.
                            set(makeUninterruptible(outputFuture).get());
                        } catch (ExecutionException e) {
                            // Set the cause of the exception as this future's
                            // exception
                            setException(e.getCause());
                        }
                    }
                }, Executors.sameThreadExecutor());
            } catch (UndeclaredThrowableException e) {
                // Set the cause of the exception as this future's exception
                setException(e.getCause());
            } catch (RuntimeException e) {
                // This exception is irrelevant in this thread, but useful for
                // the
                // client
                setException(e);
            } catch (Error e) {
                // This seems evil, but the client needs to know an error
                // occured and
                // the error needs to be propagated ASAP.
                setException(e);
                throw e;
            }
        }
    }

    /**
     * A checked future that uses a function to map from exceptions to the
     * appropriate checked type.
     */
    private static class MappingCheckedFuture<T, E extends Exception> extends
            AbstractCheckedFuture<T, E> {

        final Function<Exception, E> mapper;

        MappingCheckedFuture(ListenableFuture<T> delegate,
                Function<Exception, E> mapper) {
            super(delegate);

            this.mapper = mapper;
        }

        @Override
        protected E mapException(Exception e) {
            return mapper.apply(e);
        }
    }

    /**
     * An adapter to turn a {@link Future} into a {@link ListenableFuture}. This
     * will wait on the future to finish, and when it completes, run the
     * listeners. This implementation will wait on the source future
     * indefinitely, so if the source future never completes, the adapter will
     * never complete either.
     * 
     * <p>
     * If the delegate future is interrupted or throws an unexpected unchecked
     * exception, the listeners will not be invoked.
     */
    private static class ListenableFutureAdapter<T> extends ForwardingFuture<T>
            implements ListenableFuture<T> {

        private static final Executor adapterExecutor = java.util.concurrent.Executors
                .newCachedThreadPool();

        // The execution list to hold our listeners.
        private final ExecutionList executionList = new ExecutionList();

        // This allows us to only start up a thread waiting on the delegate
        // future
        // when the first listener is added.
        private final AtomicBoolean hasListeners = new AtomicBoolean(false);

        // The delegate future.
        private final Future<T> delegate;

        ListenableFutureAdapter(final Future<T> delegate) {
            this.delegate = delegate;
        }

        @Override
        protected Future<T> delegate() {
            return delegate;
        }

        /* @Override */
        public void addListener(Runnable listener, Executor exec) {

            // When a listener is first added, we run a task that will wait for
            // the delegate to finish, and when it is done will run the
            // listeners.
            if (!hasListeners.get() && hasListeners.compareAndSet(false, true)) {
                adapterExecutor.execute(new Runnable() {
                    /* @Override */
                    public void run() {
                        try {
                            delegate.get();
                        } catch (CancellationException e) {
                            // The task was cancelled, so it is done, run the
                            // listeners.
                        } catch (InterruptedException e) {
                            // This thread was interrupted. This should never
                            // happen, so we
                            // throw an IllegalStateException.
                            throw new IllegalStateException(
                                    "Adapter thread interrupted!", e);
                        } catch (ExecutionException e) {
                            // The task caused an exception, so it is done, run
                            // the listeners.
                        }
                        executionList.run();
                    }
                });
            }
            executionList.add(listener, exec);
        }
    }
}
